#include "EventLoop.h"
#include "Channel.h"
#include "EPollPoller.h"
#include "Logger.h"
#include "Poller.h"

#include <sys/eventfd.h>
#include <unistd.h>
#include <fcntl.h>
#include <errno.h>
#include <memory>

// 防止一个线程创建多个EventLoop，初始化为空，创建后线程内部指向创建的EvenLoop对象，不会创建多个
__thread EventLoop* t_loopInThisThread = nullptr;  // __thread 也表明线程局部存储 thread_loacl

// 定义Poller IO复用接口的超时时间，10000毫秒（10秒）
const int kPollTimeMs = 10000;

// 创建wakeupfd，用来notify唤醒subReactor处理新来的Channel，mainReactor是通过轮询的方式
int createEventfd()
{
    int evtfd = ::eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC);
    if(eventfd < 0)
    {
        LOG_FATAL("eventfd error: %d\n", errno);
    }
    return evtfd;
}

EventLoop::EventLoop()
    : looping_(false)
    , quit_(false)
    , callingPendingFunctors_(false)
    , threadId_(CurrentThread::tid())
    , poller_(Poller::newDefaultPoller(this))
    , wakeupFd_(createEventfd())
    , wakeupChannel_(new Channel(this, wakeupFd_))  // 只是注册了fd，没有设置感兴趣的事件
{
    LOG_DEBUG("EventLoop created %p in thread %d.\n", this, threadId_);
    if(t_loopInThisThread)
    {
        // 当前线程已经有一个EventLoop了
        LOG_FATAL("Another EventLoop %p exists in this thread %d.\n", t_loopInThisThread, threadId_);
    }
    else
    {
        t_loopInThisThread = this;
    }

    // 设置wakeupFd的事件类型以及发生事件后的回调操作
    wakeupChannel_->setReadCallback(std::bind(&EventLoop::handleRead, this));
    // 每一个eventLoop都将监听wakeupChannel的EPOLLIN读事件了
    wakeupChannel_->enableReading();
}

EventLoop::~EventLoop()
{
    wakeupChannel_->disableAll();
    wakeupChannel_->remove();
    ::close(wakeupFd_);
    t_loopInThisThread = nullptr;
}

// 开启事件循环
void EventLoop::loop()
{
    looping_ = true;
    quit_ = false;

    LOG_INFO("EventLoop %p start looping. \n", this);

    while (!quit_)
    {
        activeChannels_.clear();
        // 监听两类fd，一种是client的fd，一种是wakeupfd
        pollReturnTime_ = poller_->poll(kPollTimeMs, &activeChannels_);

        for(Channel* channel: activeChannels_)
        {
            // Poller能够监听哪些channel发生事件，然后上报EventLoop，EventLoop通知channel处理相应事件
            channel->handleEvent(pollReturnTime_);
        }

        // 执行当前EventLoop事件循环需要处理的回调操作
        /**
         * IO线程 mainReactor 主要就干accept，接收新用户的连接，返回和客户端通信用的fd(打包到channel里)
         * 已连接用户的channel得分发给subReactor
         * mainReactor 事先注册一个回调cb（需要subReactor执行）
         * mainReactor wakeup subReactor后，执行下面的方法，执行之前mainReactor注册的cb操作
        */
        doPendingFunctors();  // 执行队列中待处理的回调函数
    }
    
    LOG_INFO("EventLoop %p stop looping.\n", this);
}

// 退出事件循环 1. loop在自己的线程中调用quit()  2. 在非loop的线程中，调用loop的quit
void EventLoop::quit()
{
    quit_ = true;
    if(!isInLoopThread())  // 如果在其它线程中，调用quit  在subLoop(worker)中调用了mainLoop的quit
    {
        wakeup();  // 从poll阻塞唤醒，再次while的时候条件不成立，break
    }
}

// 在当前loop中执行cb
void EventLoop::runInLoop(Functor cb)
{
    if(isInLoopThread())  // 在当前loop线程中执行cb
    {
        cb();
    }
    else  // 在非当前loop线程中执行cb，就需要唤醒loop所在线程，执行cb
    {
        queueInLoop(std::move(cb));
    }
}

// 把cb放到队列中，唤醒loop所在的线程，执行cb（不在当前的loop执行）
void EventLoop::queueInLoop(Functor cb)
{
    {
        std::unique_lock<std::mutex> lock(mutex_);
        pendingFunctors_.emplace_back(cb);
    }

    // 唤醒相应的，需要执行上面回调操作的loop的线程
    // callingPendingFunctors_：当前loop正在执行回调，loop()中在doPendingFunctors()这句，但是上面又往队列里加了新的的回调，doPendingFunctors()执行完loop会阻塞，所以要唤醒去执行新的callback
    if(!isInLoopThread() || callingPendingFunctors_)
    {
        wakeup();  // 唤醒loop所在线程
    }
}

void EventLoop::handleRead()
{
    uint64_t one = 1;
    ssize_t n = read(wakeupFd_, &one, sizeof one);
    if(n != sizeof one)
    {
        LOG_ERROR("EventLoop::handleRead() reads %ld bytes instead of 8", n);
    }
}

// 用来唤醒loop所在的线程 向wakeupfd_写一个数据，wakeupChannel就发生读事件，当前loop线程就会被唤醒
void EventLoop::wakeup()
{
    uint64_t one = 1;
    ssize_t n = write(wakeupFd_, &one, sizeof one);
    if(n != sizeof one)
    {
        LOG_ERROR("EventLoop::wakeup() writes %lu bytes instead of 8. \n", n);
    }
}

void EventLoop::updateChannel(Channel* channel)
{
    poller_->updateChannel(channel);
}

void EventLoop::removeChannel(Channel* channel)
{
    poller_->removeChannel(channel);
}

bool EventLoop::hasChannel(Channel* channel)
{
    return poller_->hasChannel(channel);
}

void EventLoop::doPendingFunctors()
{
    // 局部vec和pendingFunctors_交换，相当于把pendingFunctors_置空了，把所有待执行的回到放到局部vec中
    // 如果待执行的回调比较多，因为要互斥访问，所以可能一直拿着锁，导致mainloop无法及时写进回调
    std::vector<Functor> functors;
    callingPendingFunctors_ = true;

    {
        std::unique_lock<std::mutex> lock(mutex_);
        functors.swap(pendingFunctors_);  // 待执行的回调没有全部执行完，但是不妨碍mainLoop继续写回调
    }

    for(const Functor& functor: functors)
        functor();  // 执行当前loop需要执行的回调操作
    
    callingPendingFunctors_ = false;
}